package com.dorm.cj.dr.demo;

import com.dorm.cj.dr.MyEvent;
import com.dorm.cj.dr.MyEventFactory;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.Executors;

/**
 * disruptor操作工具类的实例工厂
 */
public class DisruptorQueueFactory {

    /**
     * // 创建"点对点模式"的操作队列，即同一事件会被一组消费者其中之一消费
     *
     * @param queueSize
     * @param multiProducer
     * @param consumers
     * @return
     */
    public static DisruptorQueue getWorkPoolQueue(int queueSize, boolean multiProducer, AbsEventConsumer<MyEvent>... consumers) {
        Disruptor disruptor = new Disruptor(new MyEventFactory(), queueSize, Executors.defaultThreadFactory(),
                multiProducer ? ProducerType.MULTI : ProducerType.SINGLE, new SleepingWaitStrategy());
        disruptor.handleEventsWithWorkerPool(consumers);
        return new DisruptorQueue(disruptor);
    }

    // 创建"发布订阅模式"的操作队列，即同一事件会被多个消费者并行消费
    public static DisruptorQueue getHandleEventsQueue(int queueSize, boolean multiProducer,
                                                      AbsEventConsumer<MyEvent>... consumers) {
        Disruptor disruptor = new Disruptor(new MyEventFactory(), queueSize, Executors.defaultThreadFactory(),
                multiProducer ? ProducerType.MULTI : ProducerType.SINGLE, new SleepingWaitStrategy());
        disruptor.handleEventsWith(consumers);//可以像责任链一样定义消费顺序 dw.handleEventsWith(A).then(B);
        return new DisruptorQueue(disruptor);
    }
}
